{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "_M1WAAa_N3MO"
   },
   "source": [
    "## 1. Introduction to Tuplex\n",
    "[Tuplex](https://tuplex.cs.brown.edu) is a novel big data analytics framework allowing to execute user-defined functions in Python at native code speeds.\n",
    "\n",
    "<img src=\"https://tuplex.cs.brown.edu/_static/img/logo.png\" width=\"128px\" style=\"float: right;\" />\n",
    "\n",
    "The following notebook allows you to run Tuplex interactively and learn key concepts on the way!\n",
    "\n",
    "\n",
    "### 1.1 Setup via pip\n",
    "Tuplex can be easily installed using the following pip command (python3.7 - python3.9). For other versions, please build Tuplex from source.\n",
    "\n",
    "```\n",
    "pip3 install tuplex\n",
    "```\n",
    "\n",
    "If you're executing this notebook from the docker container `tuplex/tuplex` - you can skip the setup, Tuplex is already installed!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "ZOobvZcVO2_H"
   },
   "source": [
    "### 1.2 Creating and configuring a context object\n",
    "All computation starts by importing the module and creating a Tuplex context objects."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "IAGEc-okO2im",
    "outputId": "cd664182-903e-4be5-be66-ebc1225dea65"
   },
   "outputs": [],
   "source": [
    "import tuplex\n",
    "\n",
    "c = tuplex.Context()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "hghmsswQPDgB"
   },
   "source": [
    "A Tuplex context object can be configured in multiple ways, to see a list of available options, simply run `.options` on an existing Context object."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "TvV34AsiPL5o",
    "outputId": "edb325d9-763b-492d-8d16-17ca08ede2e6"
   },
   "outputs": [],
   "source": [
    "c.options()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "jwb4-lghPj1K"
   },
   "source": [
    "In the following, let's walk over a couple general, important options to tune Tuplex:\n",
    "\n",
    "- `executorCount` determines how many threads in addition to the main-thread Tuplex should use. I.e., when set to 3 Tuplex uses 4-way parallelism.\n",
    "- `driverMemory`/`executorMemory` determines how much memory the main thread/executor threads should use during computation. Can be configured using either a number in bytes or by passing in a string like `2G` for 2 Gibibytes. E.g., when setting driverMemory=1G and executorMemory=1g, exeutorCount=5 Tuplex will use a total of 6 Gibibytes.\n",
    "- `partitionSize` Each Task corresponds to a block of memory. partitionsize determines the task size. \n",
    "\n",
    "Other settings which change the behavior of the compiler or allow to enable/disable certain optimizations are discussed later in this Intro series."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "FtCDIIYrPM0g"
   },
   "outputs": [],
   "source": [
    "# create a tiny, single-threaded context\n",
    "c2 = tuplex.Context(executorCount=0, driverMemory='400KB')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "cZRpoc-3USWM"
   },
   "outputs": [],
   "source": [
    "# create context using options from a json dictionary\n",
    "conf = {'executorCount' : 0, \n",
    "        'driverMemory': '400KB'}\n",
    "        \n",
    "c3 = tuplex.Context(conf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "QhWlg61FUhQD"
   },
   "outputs": [],
   "source": [
    "# configure context object via YAML file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "ZPmkZ97KUjz6",
    "outputId": "49a5172c-565d-4a47-9701-d1e2aea1fc06"
   },
   "outputs": [],
   "source": [
    "%%file config.yaml\n",
    "\n",
    "# this creates a new config file in the current directory which can be passed to tuplex as well.\n",
    "tuplex:\n",
    "    -   driverMemory: 400KB\n",
    "    -   executorCount: 0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "_g0RrGMLUyLe"
   },
   "outputs": [],
   "source": [
    "c4 = tuplex.Context('config.yaml')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "gA_XssXUT8U0"
   },
   "outputs": [],
   "source": [
    "# release small contexts\n",
    "del c2 \n",
    "del c3\n",
    "del c4"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "mr1OkU8TU8Ai"
   },
   "source": [
    "## 1.3 Writing your first pipeline in Tuplex\n",
    "Pipelines in Tuplex are composed by calling operations on datasets. I.e., an operator transforms one dataset into another.\n",
    "\n",
    "In order to pass in data to a pipeline, a source is declared using one of the operations available from the Tuplex context object. Examples for such operations are\n",
    "\n",
    "- `parallelize` allows to pass Python objects as source to Tuplex\n",
    "- `csv` allows to pass one or more files (using a UNIX wildcard pattern) to Tuplex\n",
    "- ..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "lWAQkWmuVgfL"
   },
   "outputs": [],
   "source": [
    "# this creates a new dataset\n",
    "ds = c.parallelize([1, 2, 3, 4])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "CF0vqddfVpfa",
    "outputId": "8329a4bc-82d2-46a6-a833-963a454f3040"
   },
   "outputs": [],
   "source": [
    "# To pass the data back as python objects, use .collect()\n",
    "ds.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "S-URoN4BV26O"
   },
   "source": [
    "Of course passing data to and from Tuplex is a first step, but it's way more interesting to work with the actual data!\n",
    "\n",
    "For this, Tuplex provides the ability to use user-defined functions (UDFs) with various operators."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "DLLdxWhVV0oN",
    "outputId": "62831ed7-2951-43e4-e62c-14523cdc8717"
   },
   "outputs": [],
   "source": [
    "# Lambda expressions can be used\n",
    "ds.map(lambda x: x * x).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "HFetVO8uWHwp",
    "outputId": "db23d2d0-e4bf-448b-f088-8c2caed2f727"
   },
   "outputs": [],
   "source": [
    "# functions as well\n",
    "def f(x):\n",
    "    return x * x + 1\n",
    "\n",
    "ds.map(f).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "fpeRBfewefPj"
   },
   "source": [
    "Besides transforming objects via functions, we can also use a UDF to filter out certain elements. I.e., to retain only the even numbers, we can use the condition `x % 2 == 0`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "2I0o5_J1emVi",
    "outputId": "abaa45bc-a958-40ba-e55b-e7129bcdec0e"
   },
   "outputs": [],
   "source": [
    "ds.filter(lambda x: x % 2 == 0).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "ZDYGCDSfe2g4"
   },
   "source": [
    "Naturally, pipelines can be composed of multiple operations. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "j52UR4dYew7D",
    "outputId": "857a319f-34f9-4e3f-da29-3adff55fd796"
   },
   "outputs": [],
   "source": [
    "g = lambda x: x % 2 == 0\n",
    "\n",
    "ds.map(f).filter(g).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "7Ilcwt4wfCw4"
   },
   "source": [
    "Each operation thereby creates (lazily) a new dataset."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "o_WQSt6we_D5",
    "outputId": "9c8ef6a1-799b-4e03-bf2f-26337db0e970"
   },
   "outputs": [],
   "source": [
    "ds_transformed = ds.map(f)\n",
    "\n",
    "ds_filtered = ds_transformed.filter(g)\n",
    "\n",
    "ds_filtered.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "aNGAhjc7ecnB"
   },
   "source": [
    "## 1.4 Columns, tuples and names\n",
    "\n",
    "In the above examples, single integers were used as input. Yet, similar to a standard relational database, it often makes sense to structure data in a tabular format. Indeed, many input formats are given in tabular form.\n",
    "\n",
    "In Tuplex, the elementary object are tuples. Each object within a tuple may be indexed via a name. To understand this better, let's start with a simple example where multiple tuples are transformed and explore the various valid syntax options Tuplex provides:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "mbgWAll2f3Af"
   },
   "outputs": [],
   "source": [
    "ds = c.parallelize([(1, 2), (3, 4), (5, 6)])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "t8CpYg98gnQ8",
    "outputId": "328bc022-5879-41ce-931f-3d592f9f8705"
   },
   "outputs": [],
   "source": [
    "# show allows to display the data in tabular form\n",
    "ds.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "8RkD164Mf3yj"
   },
   "source": [
    "Individual columns/elements of the underlying tuples can be accessed either via multiple parameters (matching number required) or the standard indexing syntax in Python:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "6IOiyte6gszp",
    "outputId": "9835aa22-53ea-463b-dfe2-f04a1765791d"
   },
   "outputs": [],
   "source": [
    "ds.map(lambda a, b: a + b).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "zQnQLk_ngs14",
    "outputId": "c9f1c862-7530-4cba-97b5-ff895a695c75"
   },
   "outputs": [],
   "source": [
    "ds.map(lambda t: t[0] + t[1]).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "ibeU3iTriOYW"
   },
   "source": [
    "Because remembering indices for tuples with a lot of elements may be cumbersome, therefore elements may be indexed via strings within UDFs as well:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "bvFJI-NYWMYO"
   },
   "outputs": [],
   "source": [
    "ds = c.parallelize([(1, 2), (3, 4), (5, 6)], columns=['first', 'second'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "zQ6ShAiyiVEy",
    "outputId": "bc447def-7f6b-48c9-8390-be91ae4c762e"
   },
   "outputs": [],
   "source": [
    "ds.map(lambda t: t['first'] + t['second']).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "LHsxlK5MixwM"
   },
   "source": [
    "When using Tuplex, both the string access and the integer based access syntax may be freely mixed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "ZSQ06p_mivgz",
    "outputId": "528d50aa-dde6-4dba-8199-9634fbb7f24c"
   },
   "outputs": [],
   "source": [
    "ds.map(lambda t: t[0] + t['second']).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "RtQiikWZi8eQ"
   },
   "source": [
    "When looking at the output of show, we see that Tuplex has no column assigned to the end-result. In the next section, we'll take a look at how to work with columns.\n",
    "Yet, there's also a convenient syntax to provide a column name using only the `map` operation which is the case when a dictionary using string-only keys is provided:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "GBVFjpSbi6JN",
    "outputId": "85d08690-040a-4a32-fb0a-904eff09b96b"
   },
   "outputs": [],
   "source": [
    "ds.map(lambda t: {'Result' : t[0] + t['second']}).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "ddGAZYG6j6up"
   },
   "source": [
    "\n",
    "### 1.5 Manipulating individual columns, adding new columns\n",
    "\n",
    "Sometimes, only individual columns need to be manipulated or new ones should be created from existing ones. Tuplex provides two helper functions for this:\n",
    "\n",
    "`mapColumn` and `withColumn`, each taking a string to identify a column and a UDF to apply.\n",
    "\n",
    "In addition, to change the association of a column to particular name/key, the Tuplex API provides a `renameColumn` operation.\n",
    "\n",
    "Let's first take a look at how individual columns can be associated with names:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "fln8Xmqx5PVl",
    "outputId": "60350dbc-2088-452b-98d9-25d576703397"
   },
   "outputs": [],
   "source": [
    "# name columns by passing in information to parallelize\n",
    "ds = c.parallelize([(1, 2), (3, 4), (5, 6)], columns=['first', 'second'])\n",
    "\n",
    "ds.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "zEnbbYL05jaR",
    "outputId": "2bee4434-3466-48bb-d83c-ada7912b4f1b"
   },
   "outputs": [],
   "source": [
    "ds = c.parallelize([(1, 2), (3, 4), (5, 6)]).renameColumn(0, 'first')\n",
    "ds.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "bZlv0c3n5q19",
    "outputId": "b0125a3e-7300-45ee-e111-e722afbfa5a8"
   },
   "outputs": [],
   "source": [
    "ds = ds.renameColumn(1, 'second').renameColumn('first', 'FIRST')\n",
    "\n",
    "ds.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "GzXR_l0d5xyN",
    "outputId": "5e79f94c-42b0-42c6-9683-eb663e29b02a"
   },
   "outputs": [],
   "source": [
    "ds.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "NJQSV3fW5zUX"
   },
   "outputs": [],
   "source": [
    "ds = ds.renameColumn(0, 'first')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "Nv-jEpN-5QPN"
   },
   "source": [
    "In a next step, we can create new columns based on all columns via `withColumn` or manipulate a single column using `mapColumn`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "8-8ZuTR4i6RH",
    "outputId": "d785ac0a-e960-488f-fb5a-18bf0d62fd5b"
   },
   "outputs": [],
   "source": [
    "ds.withColumn('third', lambda a, b: a + b).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "-UmYPS18kePQ",
    "outputId": "6a634f75-709e-4c75-d2df-274b27838251"
   },
   "outputs": [],
   "source": [
    "ds.mapColumn('first', lambda x: x - 1).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "CGbXRmfXk0R3"
   },
   "source": [
    "Lastly, sometimes only a subset of columns is required. For this, Tuplex provides a `selectColumns` operation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "5vhtuK7Wkz2X",
    "outputId": "5d6d330c-8e97-49a1-9e5b-1384a2386e9a"
   },
   "outputs": [],
   "source": [
    "ds.withColumn('third', lambda a, b: a + b).selectColumns(['first', 'third']).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "Su5n80pfUJKt"
   },
   "source": [
    "This notebook showed some basic interactions and manipulations of datasets. However, the data diplayed here is rather small and the benefits of a compiling framework will shine when dealing with larger amounts of data. In the [next part II](02_Working_with_files.ipynb), we'll therefore learn more about how to work efficiently with files in Tuplex!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "7_iruiYiiWmd"
   },
   "source": [
    "(c) 2017 - 2022 Tuplex team"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "wx5lb9RQXqvf"
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "colab": {
   "collapsed_sections": [],
   "provenance": []
  },
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}
